#include "tcp_io_loop.hh"
#include "parameter.hh"
#include "network_selector.hh"
#include "../logger/writer.hh"
#include "../util/os_util.hh"
#include <functional>

using namespace kratos::corelib;

namespace kratos {
namespace network {
TcpIoLoop::TcpIoLoop()
    : kloop_(nullptr), sessionBuffer_(new char[NETWORK_IOLOOP_RECV_BUFFER_MAX_SIZE]) {
}

TcpIoLoop::~TcpIoLoop() {
    for (auto it : chanMap_) {
        knet_channel_ref_close(it.second);
    }
    if (kloop_) {
        knet_loop_destroy(kloop_);
    }
    delete[] sessionBuffer_;
    sessionBuffer_ = nullptr;
    kloop_ = nullptr;
}

void TcpIoLoop::start(NetworkSelector * loop) {
    NetIoLoop::start(loop);
}

void TcpIoLoop::update() {
    knet_loop_run_once(kloop_);
}

std::uint64_t TcpIoLoop::onNewAcceptor(const std::string& ip, const uint16_t port,
    std::size_t netBufSize) {
    if (!kloop_) {
        kloop_ = knet_loop_create();
    }
    auto channel = knet_loop_create_channel(kloop_, NETWORK_IOLOOP_SEND_LIST_MAX_COUNT, netBufSize);
    knet_channel_ref_set_cb(channel, &TcpIoLoop::acceptor_cb_);
    knet_channel_ref_set_ptr(channel, this);
    auto error = knet_channel_ref_accept(channel, ip.c_str(), port, NETWORK_IOLOOP_SEND_LIST_MAX_COUNT);
    if (error != error_ok) {
        return 0;
    }
    auto channelID = knet_channel_ref_get_uuid(channel);
    chanMap_[channelID] = channel;
    return channelID;
}

std::uint64_t TcpIoLoop::onNewConnector(const std::string& ip, const uint16_t port,
    std::size_t netBufSize) {
    if (!kloop_) {
        kloop_ = knet_loop_create();
    }
    if (!kloop_) {
        return 0;
    }
    auto channel = knet_loop_create_channel(kloop_, NETWORK_IOLOOP_SEND_LIST_MAX_COUNT, netBufSize);
    auto realIp = kratos::corelib::OS::getHostIP(ip);
    auto error = knet_channel_ref_connect(channel, realIp.c_str(), port, 0);
    if (error != error_ok) {
        return 0;
    }
    knet_channel_ref_set_cb(channel, &TcpIoLoop::connector_cb_);
    knet_channel_ref_set_ptr(channel, this);
    auto channelID = knet_channel_ref_get_uuid(channel);
    chanMap_[channelID] = channel;
    return channelID;
}

void TcpIoLoop::onClose(std::uint64_t channelID) {
    if (!kloop_) {
        kloop_ = knet_loop_create();
    }
    auto it = chanMap_.find(channelID);
    if (it == chanMap_.end()) {
        return;
    }
    knet_channel_ref_close(it->second);
}

void TcpIoLoop::onSend(std::uint64_t channelID, const char* data, std::uint32_t size) {
    if (!kloop_) {
        kloop_ = knet_loop_create();
    }
    auto it = chanMap_.find(channelID);
    if (it == chanMap_.end()) {
        return;
    }    
    auto stream = knet_channel_ref_get_stream(it->second);
    auto error = knet_stream_push(stream, data, size);
    if (error != error_ok) {
        knet_channel_ref_close(it->second);
        return;
    }
}

void TcpIoLoop::onAccepted_(kchannel_ref_t* channel) {    
    if (!kloop_) {
        kloop_ = knet_loop_create();
    }
    auto channelID = knet_channel_ref_get_uuid(channel);    
    chanMap_[channelID] = channel;
    onAccepted(channelID);
}

void TcpIoLoop::onConnected_(kchannel_ref_t * channel) {
    if (!kloop_) {
        kloop_ = knet_loop_create();
    }
    auto channelID = knet_channel_ref_get_uuid(channel);    
    chanMap_[channelID] = channel;
    onConnected(channelID);
}

void TcpIoLoop::onClose_(kchannel_ref_t * channel) {
    if (!kloop_) {
        kloop_ = knet_loop_create();
    }
    auto channelID = knet_channel_ref_get_uuid(channel);
    auto it = chanMap_.find(channelID);
    if (it == chanMap_.end()) {
        return;
    }
    chanMap_.erase(it);
    onClosed(channelID);
}

void TcpIoLoop::onReceive_(kchannel_ref_t * channel) {
    auto channelID = knet_channel_ref_get_uuid(channel);    
    auto stream = knet_channel_ref_get_stream(channel);
    auto bytes = knet_stream_available(stream);
    while (bytes > 0) {
        auto readBytes = NETWORK_IOLOOP_RECV_BUFFER_MAX_SIZE > bytes ?
            bytes : NETWORK_IOLOOP_RECV_BUFFER_MAX_SIZE;
        auto error = knet_stream_pop(stream, sessionBuffer_, readBytes);
        if (error != error_ok) {
            knet_channel_ref_close(channel);
            return;
        }
        onReceived(channelID, sessionBuffer_, readBytes);
        bytes = knet_stream_available(stream);
    }
}

void TcpIoLoop::acceptor_cb_(kchannel_ref_t * channel, knet_channel_cb_event_e e) {
    auto tcpLoop = reinterpret_cast<TcpIoLoop*>(knet_channel_ref_get_ptr(channel));
    if (e & channel_cb_event_accept) {
        knet_channel_ref_set_cb(channel, &TcpIoLoop::channel_cb_);
        tcpLoop->onAccepted_(channel);
    }
}

void TcpIoLoop::connector_cb_(kchannel_ref_t * channel, knet_channel_cb_event_e e) {
    auto tcpLoop = reinterpret_cast<TcpIoLoop*>(knet_channel_ref_get_ptr(channel));
    if (e & channel_cb_event_connect) {
        knet_channel_ref_set_cb(channel, &TcpIoLoop::channel_cb_);
        tcpLoop->onConnected_(channel);
    } else {
        channel_cb_(channel, e);
    }
}

void TcpIoLoop::channel_cb_(kchannel_ref_t * channel, knet_channel_cb_event_e e) {
    auto tcpLoop = reinterpret_cast<TcpIoLoop*>(knet_channel_ref_get_ptr(channel));
    if (e & channel_cb_event_recv) {
        tcpLoop->onReceive_(channel);
    } else if (e & channel_cb_event_close) {
        tcpLoop->onClose_(channel);
    }
}

}
}

